package com.shujia.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}

object Demo01WordCount {
  def main(args: Array[String]): Unit = {

    // 创建SparkSession 从Spark 2.0引入的新入口
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo01WordCount")
      .master("local")
      .getOrCreate()

    // 通过SparkSession获取SparkContext
    val sc: SparkContext = spark.sparkContext
    // SparkContext读取文件的方式
    val lineRDD: RDD[String] = sc.textFile("Spark/data/words.txt")
    lineRDD.foreach(println)

    // 通过SparkSession读取文件
    val lineDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "|")
      .schema("words String")
      .load("Spark/data/words.txt")

    // 打印
    lineDF.show() // 默认以截断的方式打印20条数据
    lineDF.show(truncate = false) // 完整展示20条数据
    lineDF.show(2, truncate = false) // 完整展示2条数据

    // 统计每个单词的数量

    // RDD的方式
    lineRDD
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKey(_ + _)
      .foreach(println)

    // DataFrame的方式

    // 第一种：SQL
    // 需要将DataFrame注册成临时视图
    lineDF.createOrReplaceTempView("lineTable")
    spark.sql(
      """
        |select  t1.word
        |        ,count(*) as cnt
        |from (
        |    select  explode(split(words,",")) as word
        |    from lineTable
        |) t1 group by t1.word
        |
        |""".stripMargin).show()

    // 导入SparkSQL提供的函数
    import org.apache.spark.sql.functions._
    // 导入SparkSQL的隐式转换
    import spark.implicits._

    // DSL的方式 (推荐)
    lineDF
      .select(explode(split($"words", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .show()


  }

}
